﻿/*
* ==============================================================================
*
* Filename: observable_socket
* Description: 
*
* Version: 1.0
* Created: 2019/12/23 16:09:36
* Compiler: Visual Studio 2019
*
* Author: szg
* Company: freelancer
*
* ==============================================================================
*/

namespace client35 {
	using Autofac;
	using common;
	using i_core;
	using System;
	using System.Collections;
	using System.Collections.Generic;
	using System.Diagnostics;
	using System.IO;
	using System.Net;
	using System.Net.Sockets;
	using System.Reactive.Linq;
	using System.Reactive.Subjects;
	using System.Text;
	using System.Threading;
	using System.Threading.Tasks;

	public class observable_socket : IDisposable {

		public bool is_connected => tcp_client != null && tcp_client.Connected;

		public IObservable<string> _messages { get; set; }

		private InterlockedBoolean is_started = new InterlockedBoolean();
		private InterlockedBoolean is_disposed = new InterlockedBoolean();
		private Subject<string> subject;
		private TcpClient tcp_client;

		private readonly i_log log;

		public observable_socket() { }
		public observable_socket(string server_ip, int? server_port, string local_ip, int? local_port) {
			try {
				if (string.IsNullOrEmpty(server_ip))
					throw new ArgumentNullException("server_ip");

				if (!server_port.HasValue)
					throw new ArgumentNullException("server_port");

				if (string.IsNullOrEmpty(local_ip))
					throw new ArgumentNullException("local_ip");

				if (!local_port.HasValue)
					throw new ArgumentNullException("local_port");

				subject = subject ?? new Subject<string>();

				log = global_vars.scope.ResolveKeyed<i_log>(log_type.log4net,
					new NamedParameter("user", "system"),
					new NamedParameter("islog", true));


				tcp_client = new TcpClient(new IPEndPoint(IPAddress.Parse(local_ip), 0)) {
					NoDelay = true,
					SendBufferSize = 8192 * 1,
					ReceiveBufferSize = 8192 * 1
				};
				tcp_client.Connect(IPAddress.Parse(server_ip), server_port.Value);

				_messages = Observable.Defer(
					() => {
						if (is_started.EnsureCalledOnce())
							return subject.AsObservable();

						_ = Task.Factory.StartNew(
							() => {
								var dataStr = string.Empty;
								var n_stream = tcp_client.GetStream();

								try {
									while (tcp_client != null && tcp_client.Connected) {
										if (tcp_client.Available <= 0) {
											Thread.Sleep(0);
											continue;
										}

										var read_size = tcp_client.Available;
										var buffer = new byte[read_size];
										n_stream.Read(buffer, 0, buffer.Length);

										var rev_msg = Encoding.GetEncoding("GB2312").GetString(buffer);
										log.Debug($"receive message:\r\n{rev_msg}");
										subject.OnNext(rev_msg);
										Thread.Sleep(0);
									}
									log.Info($"socket has complete");
								} catch (ObjectDisposedException) {
									subject.OnCompleted();
								} catch (TaskCanceledException) {
									subject.OnCompleted();
								} catch (Exception ex) {
									log.Error("Unexpected Error reading from stream", ex);
									subject.OnError(ex);
								}
							});
						return subject.AsObservable();
					});
			} catch (Exception ex) {
				log.Error("socket error", ex);
			}
		}

		protected async Task SendAsync(string message) {
			if (tcp_client == null)
				throw new SocketException();

			try {
				log.Debug($"send message:\r\n{message}");
				var bytes = Encoding.GetEncoding("GB2312").GetBytes(message);
				var stream = tcp_client.GetStream();
				stream.Write(bytes, 0, bytes.Length);
			} catch {
				Dispose();
				throw;
			}
		}

		public void Dispose() {
			Dispose(true);
			GC.SuppressFinalize(this);
		}

		protected virtual void Dispose(bool disposing) {
			if (is_disposed != null && !is_disposed.EnsureCalledOnce()) {
				if (disposing) {
					log.Debug($"Disposing {GetType()} (disposing:{disposing})");
				}
				try {
					if (tcp_client != null && tcp_client.Connected) {
						tcp_client.Client.Shutdown(SocketShutdown.Both);

						var disconnectEventArgs = new SocketAsyncEventArgs();
						disconnectEventArgs.DisconnectReuseSocket = true;
						tcp_client.Client.DisconnectAsync(disconnectEventArgs);

						tcp_client.Close();
					}
				} catch (Exception ex) {
					log.Warn($"dispose socket error", ex);
				}
			}
		}
	}
}
